package cn.cuilan.common.client;

import cn.cuilan.common.core.AbstractProtocolConnector;
import cn.cuilan.common.entity.ProtocolMessage;
import cn.cuilan.common.pipeline.ProtocolChannelInitializer;
import cn.cuilan.common.pipeline.ReconnectOnCloseListener;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * 私有协议客户端
 *
 * @author zhang.yan
 */
public class ProtocolMessageClient<T extends ProtocolMessage>
        extends AbstractProtocolConnector<ClientConfiguration, Bootstrap, T> {

    /**
     * 重连监听器
     */
    private ReconnectOnCloseListener<T> reconnectOnCloseListener;

    public ProtocolMessageClient(SocketAddress socketAddress) {
        this(socketAddress, ClientConfiguration.getDefault());
    }

    public ProtocolMessageClient(SocketAddress socketAddress, ClientConfiguration configuration) {
        super(configuration);
        setSocketAddress(socketAddress);
    }

    @Override
    protected Bootstrap createBootstrap() {
        final Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(getBossEventLoopGroup())
                .channel(NioSocketChannel.class)
                .remoteAddress(getSocketAddress())
                .handler(new ProtocolChannelInitializer<>(
                        getConfiguration(),
                        getConfigurer(),
                        getWorkerEventLoopGroup(),
                        getMessageHandler()));
        configureBootstrap(bootstrap);
        bootstrap.validate();

        reconnectOnCloseListener = new ReconnectOnCloseListener<>(this,
                getConfiguration().getReconnectInterval(),
                getBossEventLoopGroup());
        return bootstrap;
    }

    // =========================================================================

    public boolean isConnected() {
        Channel channel = getChannel();
        return channel != null && channel.isActive();
    }

    /**
     * 同步连接至远程地址
     */
    public ChannelFuture connect() throws InterruptedException {
        final Channel channel = connectAsync().sync().channel();
        assert (channel != null) : "Channel must be set";
        setChannel(channel);
        return channel.closeFuture();
    }

    /**
     * 同步连接至远程地址
     *
     * @param host 远程地址
     * @param port 端口
     */
    public ChannelFuture connect(String host, int port) throws InterruptedException {
        return connect(new InetSocketAddress(host, port));
    }

    /**
     * 同步连接至远程地址
     *
     * @param serverAddress 远程地址
     */
    public ChannelFuture connect(SocketAddress serverAddress) throws InterruptedException {
        setSocketAddress(serverAddress);
        return connect().sync();
    }

    /**
     * 异步连接至远程地址
     */
    public ChannelFuture connectAsync() {
        logger.debug("Connecting to {}", getSocketAddress());
        final Bootstrap b = getBootstrap();
        reconnectOnCloseListener.requestReconnect();
        final ChannelFuture connectFuture = b.connect();
        // 添加监听器
        connectFuture.addListener(connFuture -> {
            if (!connectFuture.isSuccess()) {
                reconnectOnCloseListener.scheduleReconnect();
                return;
            }
            Channel channel = connectFuture.channel();
            logger.debug("Client is connected to {}", channel.remoteAddress());
            setChannel(channel);
            channel.closeFuture().addListener(reconnectOnCloseListener);
        });
        return connectFuture;
    }

    // =========================================================================

    /**
     * 同步断开连接
     */
    public void disconnect() throws InterruptedException {
        final ChannelFuture disconnectFuture = disconnectAsync();
        if (disconnectFuture != null) {
            disconnectFuture.await();
        }
    }

    /**
     * 异步断开连接
     */
    public ChannelFuture disconnectAsync() {
        reconnectOnCloseListener.requestDisconnect();
        final Channel channel = getChannel();
        if (channel != null) {
            final SocketAddress socketAddress = getSocketAddress();
            logger.info("Closing connection to {}", socketAddress);
            return channel.close();
        } else {
            return null;
        }
    }

    // =========================================================================

    /**
     * 同步发送消息，加超时等待
     *
     * @param protocolMessage 私有协议消息
     * @param timeout         超时时间
     * @param timeUnit        时间单位
     */
    public void send(ProtocolMessage protocolMessage, long timeout, TimeUnit timeUnit) throws InterruptedException {
        sendAsync(protocolMessage).sync().await(timeout, timeUnit);
    }

    /**
     * 同步发送消息
     *
     * @param protocolMessage 私有协议消息
     */
    public void send(ProtocolMessage protocolMessage) throws InterruptedException {
        sendAsync(protocolMessage).sync().await();
    }

    /**
     * 异步发送消息
     *
     * @param protocolMessage 私有协议消息
     */
    public ChannelFuture sendAsync(ProtocolMessage protocolMessage) {
        Channel channel = getChannel();
        if (channel == null) {
            throw new IllegalStateException("Channel is not opened");
        }
        if (!channel.isWritable()) {
            throw new IllegalStateException("Channel is not writable");
        }
        return channel.writeAndFlush(protocolMessage);
    }

}
